package com.yj.rocketmq.producer;


import com.yj.core.exception.ServiceException;
import com.yj.rocketmq.RocketMqConstants;
import com.yj.rocketmq.RocketMqExceptionMsg;
import com.yj.rocketmq.listener.LocalTransactionListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.Optional;
import java.util.concurrent.*;

/**
 * rocketMq 生成者
 *
 * @author Xie
 * @date 19-3-19
 */
@Slf4j
public abstract class AbstractMqTransProducer {

    @Autowired
    LocalTransactionListener localTransactionListener;

    /**
     * 生成对象
     */
    private TransactionMQProducer producer = null;

    /**
     * 返回生产者对象
     *
     * @return
     */
    public TransactionMQProducer getProducer() {
        if (producer == null) {
            log.error("{}DefaultMQProducer handler is null,dose not send message", RocketMqConstants.LOG_PREFIX);
            throw new ServiceException(RocketMqExceptionMsg.PRODUCER_HANDLE_NULL);
        }
        return producer;
    }

    /**
     * 初始化
     */
    @PostConstruct
    public void initMQProducer() {
        //检查是否启用
        if (getProducerEnable() == null || !getProducerEnable()) {
            log.warn("{}未启用TransactionMQProducer,无需初始化", RocketMqConstants.LOG_PREFIX);
            return;
        }
        //检查配置参数
        if (StringUtils.isAnyBlank(getNameSrv(), getProducerId())) {
            log.error("{} 请正确配置 TransactionMQProducer参数", RocketMqConstants.LOG_PREFIX);
            throw new ServiceException(RocketMqExceptionMsg.PRODUCER_PARAM_CFG_INVALID);
        }

        //消息发送对象
        //producer = new TransactionMQProducer(getProducerId());
        producer = new TransactionMQProducer("SINOSUN_GROUP");
        producer.setTransactionListener(localTransactionListener);
        ExecutorService executorService = new ThreadPoolExecutor(10, 100, 100, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(2000), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName("client-transaction-msg-check-thread");
                return thread;
            }
        });
        producer.setNamesrvAddr(getNameSrv());
        producer.setRetryTimesWhenSendFailed(Integer.valueOf(getMaxReproducerTimes()).intValue());
        producer.setExecutorService(executorService);
        producer.setInstanceName(getProducerId());
        try {
            producer.start();
            log.warn("{}producerId={} TransactionMQProducer开启成功!", RocketMqConstants.LOG_PREFIX, getProducer());
        } catch (MQClientException e) {
            e.printStackTrace();
            log.error("{}TransactionMQProducer 启动发生异常，异常信息={}", RocketMqConstants.LOG_PREFIX, e);
            throw new ServiceException(RocketMqExceptionMsg.PRODUCER_START_FAILED);
        }
    }


    /**
     * 在应用退出前, 销毁 Producer 对象
     */
    @PreDestroy
    public void shutDownProducer() {
        if (getProducerEnable() == null || !getProducerEnable() || producer == null) {
            log.warn("{}TransactionMQProducer，无需关闭", RocketMqConstants.LOG_PREFIX);
            return;
        }
        Optional.ofNullable(producer).ifPresent(defaultMQProducer -> {
            log.info("{}在应用退出前，销毁 TransactionMQProducer　对象", RocketMqConstants.LOG_PREFIX);
            defaultMQProducer.shutdown();
        });
    }

    /**
     * Producer　Id
     *
     * @return
     */
    protected abstract String getProducerId();

    /**
     * mq 地址
     *
     * @return
     */
    protected abstract String getNameSrv();

    /**
     * 重试次数
     *
     * @return
     */
    protected abstract String getMaxReproducerTimes();

    /**
     * 启用标志
     *
     * @return
     */
    protected abstract Boolean getProducerEnable();
}
